package com.atguigu.gmall.realtime.app

import java.lang
import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.DauInfo
import com.atguigu.gmall.realtime.util.{MyEsUtil, MyKafkaUtil, OffsetManager, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

object DauApp {

  //1、要从日志中择出访问信息，（如果只是app的话，可以去应用的启动日志 如果还有页面或者链接 取访问的第一个页面的访问日志 ）     根据last_page_id 判断 如果为空 则表示该用户本次访问的第一个页面
  //2、 对多次访问的首页日志，进行去重，redis
  //3、写入到es 中
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("dau_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    val topic = "ODS_BASE_LOG"
    val groupId = "dau_group"
    // 抓取数据
    //根据redis中存储的偏移量 ， 从kafka中取值
    val offsetMapForKafka: Map[TopicPartition, Long] = OffsetManager.getOffset(topic, groupId)
    var inputDstream: InputDStream[ConsumerRecord[String, String]] = null
    if (offsetMapForKafka != null && offsetMapForKafka.size > 0) {
      inputDstream = MyKafkaUtil.getKafkaStream(topic, ssc, offsetMapForKafka, groupId)
    } else {
      inputDstream = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)
    }
    //从流中提取偏移量结束点
    var offsetRanges: Array[OffsetRange] = null //dr 启动时一次性声明 driver变量
    val inputByOffsetDstream: DStream[ConsumerRecord[String, String]] = inputDstream.transform { rdd =>
      //dr  周期性执行
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
    }




    // inputDstream.map(_.value()).print(1000)

    //1、要从日志中择出访问信息，（如果只是app的话，可以去应用的启动日志 如果还有页面或者链接 取访问的第一个页面的访问日志 ）
    // 根据last_page_id 判断 如果为空 则表示该用户本次访问的第一个页面   //筛选
    //1.1 为了筛选操作方便 首先可以把字符串转为方便处理json对象
    // json string==>1、专用的结构化对象： bean或case class 2  通用的结构化对象 map 或者jsonObject
    val jsonDStream: DStream[JSONObject] = inputByOffsetDstream.map { record =>
      val jsonString: String = record.value()
      val jSONObject: JSONObject = JSON.parseObject(jsonString)
      //补充对日期加工  ts -> hr ts->yyyy-MM-dd
      val ts: lang.Long = jSONObject.getLong("ts")

      val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
      val dateHrString: String = dateFormat.format(new Date(ts))
      val dataHrArr: Array[String] = dateHrString.split(" ")
      val dt: String = dataHrArr(0)
      val hr: String = dataHrArr(1)
      jSONObject.put("dt", dt)
      jSONObject.put("hr", hr)
      jSONObject

    }
    //1.2 根据last_page_id 判断 如果为空 则表示该用户本次访问的第一个页面
    // 筛选
    val firstDStream: DStream[JSONObject] = jsonDStream.filter { jSONObject =>
      var ifFirst = false
      val pageObj: JSONObject = jSONObject.getJSONObject("page")
      if (pageObj != null) {
        val lastPageId: String = pageObj.getString("last_page_id")
        if (lastPageId != null && lastPageId.size > 0) {
          ifFirst = false
        } else {
          ifFirst = true //只有包含访问的日志 且该访问没有上一页  则视为首页保留
        }
      }
      ifFirst
    }
    // firstDStream.print(1000)

    // 去重 //利用redis 进行去重  记录当日访问过的用户清单
    //type ?   set   key ?  dau:2021-10-29   value? mid   field/score ?  无   写入api?  sadd
    // 读取api ?  sismember     失效时间？ 2 天
    // map 不会减少数据
    // 1  有没有并发问题    sismeber +sadd ==> sadd
    // 2  性能能不能再优化一点

    // val jedis: Jedis = RedisUtil.getJedisClient  //jedis driver   // 不行 会抛序列化问题

    //    val filteredDStream: DStream[JSONObject] = firstDStream.filter { jsonObj =>
    //      val commonJSONObj: JSONObject = jsonObj.getJSONObject("common")    //executor
    //      val mid: String = commonJSONObj.getString("mid")
    //      val dt: String = jsonObj.getString("dt")
    //      val key = "dau:" + dt
    //     //?
    //      val isNonExistsFlag: lang.Long = jedis.sadd(key, mid)  // sadd  既可以进行判断 有可以写入  原子性命令
    //      if(isNonExistsFlag==1L){
    //          true
    //      }else{
    //         false
    //      }
    //    }
    // jedis.close()
    val dauDstream: DStream[JSONObject] = firstDStream.mapPartitions { jsonObjItr => //每分区 每批次的数据
      val jedis: Jedis = RedisUtil.getJedisClient //分区共用
    val targerList: ListBuffer[JSONObject] = new ListBuffer[JSONObject]
      val originList: List[JSONObject] = jsonObjItr.toList
      println("该分区共：" + originList.size + "条")
      for (jsonObj <- originList) { //每条数据
        val commonJSONObj: JSONObject = jsonObj.getJSONObject("common") //executor
      val mid: String = commonJSONObj.getString("mid")
        val dt: String = jsonObj.getString("dt")
        val key = "dau:" + dt

        val isNonExistsFlag: lang.Long = jedis.sadd(key, mid) // sadd  既可以进行判断 有可以写入  原子性命令
        if (isNonExistsFlag == 1L) {
          targerList.append(jsonObj)
        }
      }
      jedis.close()
      println("该分区筛选后 共：" + targerList.size + "条")
      targerList.toIterator
    }
    // dr   // 启动时执行一次
    //    dauDstream.transform{rdd=>
    //      // dr ? driver     //  周期性
    //      rdd.map{
    //        null  //ex
    //      }
    //      null
    //    }

    //向olap中写入数据
    dauDstream.foreachRDD { rdd =>

      rdd.foreachPartition { dauItr =>
        //要写入的索引前缀gmall0609_dau_info
        val list: List[JSONObject] = dauItr.toList
        if (list.size >= 1) {
          // 批量写入数据
          val dauList: List[(String, DauInfo)] = list.map { jsonObj =>
            val commonObj: JSONObject = jsonObj.getJSONObject("common")
            val mid: String = commonObj.getString("mid")
            val uid: String = commonObj.getString("uid")
            val ar: String = commonObj.getString("ar")
            val ch: String = commonObj.getString("ch")
            val vc: String = commonObj.getString("vc")
            val dt: String = jsonObj.getString("dt")
            val hr: String = jsonObj.getString("hr")
            val ts: Long = jsonObj.getLong("ts")
            val dauInfo = DauInfo(mid, uid, ar, ch, vc, dt, hr, ts)
            // 要根据业务选择唯一性标识 ： 保证幂等性
            (mid, dauInfo)

          }
          val firstDauInfo: DauInfo = dauList(0) _2
          val indexName: String = "gmall0609_dau_info_" + firstDauInfo.dt
          MyEsUtil.bulkSave(dauList, indexName)

          println("要存储的数据：")
          println(list.mkString("\n"))
        }
        //提交偏移量A ？ dr? ex?  ex
      }
      //提交偏移B ？ dr? ex?  dr
      OffsetManager.saveOffset(topic, groupId, offsetRanges)
    }

    //


    ssc.start()
    ssc.awaitTermination()

  }

}
